package com.rocketmq.controller.produce;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.List;

/**
 * 订单系统生产者
 */
@Controller
@RequestMapping("/order")
public class OrderSystemProducer {

    private static final Logger logger = LoggerFactory.getLogger(OrderSystemProducer.class);

    @RequestMapping("/produce")
    @ResponseBody
    public String produceOrder() {
        SendResult sendResult = new SendResult();
        try {
            DefaultMQProducer producer = new DefaultMQProducer("OrderSystemProducer");
            // 元数据连接地址
            producer.setNamesrvAddr("100.85.217.111:8200;100.93.3.194:8200");
            producer.start();

            // 订单消息标识
            String[] tags = new String[]{"CommonOrderTag", "ScheduledOrderTag"};
            String orderId = "order0";

            // 生产订单消息
            Message msg = new Message("OrderTopic", tags[0], "KEY0",
                    ("common order message").getBytes(StandardCharsets.UTF_8));
            sendResult = producer.send(msg, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    String orderId = (String) arg;
                    int index = Math.abs(orderId.hashCode() % mqs.size());
                    return mqs.get(index);
                }
            }, orderId);

            // 延时发送生产订单消息
            // 延时消息投递时间戳，该消息60秒后投递
            final long deliverTimestamp = Instant.now().plusSeconds(60).toEpochMilli();
            // 创建消息对象
            Message scheduledMsg = new Message("OrderTopic",
                    tags[1],
                    "KEY1",
                    "scheduled order message".getBytes(StandardCharsets.UTF_8));
            // 设置消息定时投递的时间戳属性
            scheduledMsg.putUserProperty("__STARTDELIVERTIME", String.valueOf(deliverTimestamp));
            producer.send(scheduledMsg);

            producer.shutdown();
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
            logger.error("send message exception {}", e);
        }
        return sendResult.toString();
    }
}
